﻿//Description:接收处理类-数据变更记录
//Copyright (c) : 通力凯顿（北京）系统集成有限公司
//Writer:Wangjh
//create Date:2023-8-9
//Rewriter:
//Rewrite Date:
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Configuration;
using System.IO;
using Newtonsoft.Json;
using DatabaseAccess.DBUtility;
using System.Net;
using BusinessRule.Common;
using System.Threading.Tasks;
using Newtonsoft.Json.Linq;
//using StackExchange.Redis;

namespace OracleTableSync
{
    public class ReceiveAndDeal
    {
        //ConnectionMultiplexer onn = new ConnectionMultiplexer();
        SyncTableStructure sts = new SyncTableStructure();
        SyncTableAllData sta = new SyncTableAllData();
        public static readonly String receivePath = ConfigurationManager.AppSettings["ReceivePath"].ToString();//接收文件路径
        public static readonly String receiveFileBak = ConfigurationManager.AppSettings["receiveFileBak"].ToString();//接收文件路径-历史
        public static readonly String receiveFileBakError = ConfigurationManager.AppSettings["receiveFileBakError"].ToString();//接收文件路径-历史，处理错误路径 
        private static readonly String tabMapFilePath = ConfigurationManager.AppSettings["tabMapFilePath"].ToString();
        private static List<JObject> tabMapTab;//字段映射关系
        private List<Task> allTaskList ;//异步的任务
        /// <summary>
        /// 执行入口
        /// </summary>
        public void Excute()
        {
            LogHelp.SetConfig();
            allTaskList = new List<Task>();
            //处理接收的表结构
            //sts.ReceiveSyncStructData();
            //处理接收的数据
            DealReceiveSyncData();
            Task.WaitAll(allTaskList.ToArray());//等待异步的逻辑
        }

        /// <summary>
        /// 处理接收到的数据，经本地测试一分钟能接收处理6W左右数据
        /// 必须严格按文件的顺序，文件内json的顺序先后执行
        /// </summary>
        /// <param name="dt"></param>
        void DealReceiveSyncData()
        {
            using (StreamReader streamReader = new StreamReader(tabMapFilePath))
            {
                string jsonText = streamReader.ReadToEnd();
                tabMapTab = JsonConvert.DeserializeObject<List<JObject>>(jsonText);
            }
            DirectoryInfo mydir = new DirectoryInfo(receivePath);
            FileInfo[] files = mydir.GetFiles().Where(item => item.Name.Contains("_SyncData_")).ToArray(); 
            foreach (FileInfo fi in files)
            {
                try
                {
                    DealSyncData(fi);
                }
                catch (Exception ex)
                {
                    LogHelp.WriteError(ex.Message, ex);
                    //处理异常
                    CommonHelper.moveFile(fi, receiveFileBakError);
                }
            }        
        }
         
        /// <summary>
       /// 处理接收到的数据
       /// </summary>
       /// <param name="dt"></param>
        void DealSyncData(FileInfo fi)
        {
            if (fi.Extension!=".json") {
                return;
            }
            List<ExcuteEntity> exs = new List<ExcuteEntity>();
            List<ExcuteEntity> exsAsync = new List<ExcuteEntity>();//异步执行的sql
            //String receiveID = Guid.NewGuid().ToString();//一次接收的id 

            using (StreamReader streamReader = new StreamReader(fi.FullName))
            {
                string jsonText = streamReader.ReadToEnd();

                List<Dictionary<String, Object>> obj = JsonConvert.DeserializeObject<List<Dictionary<String, Object>>>(jsonText);                
                foreach (Dictionary<String, Object> eachData in obj) {
                    if (eachData["type"].ToString() != "syncData")
                        continue;
                    
                    String tableName = eachData["tableName"].ToString();
                    String orgDb = eachData["orgDb"].ToString();
                    String toDb = eachData["toDb"].ToString();
                    //插入接收表
                    exsAsync.Add(new ExcuteEntity()
                    {
                        TableName = "Sync_ReceiveRecord",
                        ExcuteType = ExcuteEntity.ExcuteTypeEnum.insert,
                        ExcuteFileds = new List<FieldEntity>() { new FieldEntity("id", Guid.NewGuid().ToString(), FieldEntity.FieldTypeEnum.Str),
                           new FieldEntity("ReceiveID", eachData["pushID"], FieldEntity.FieldTypeEnum.Str),
                           new FieldEntity("TableChangeRecordID", eachData["tableChangeRecordID"].ToString(), FieldEntity.FieldTypeEnum.Str),
                           new FieldEntity("TableName", eachData["tableName"].ToString(), FieldEntity.FieldTypeEnum.Str),
                           new FieldEntity("OldValue", eachData["oldValue"].ToString(), FieldEntity.FieldTypeEnum.Clob),
                           new FieldEntity("NewValue", eachData["newValue"].ToString(), FieldEntity.FieldTypeEnum.Clob),
                           new FieldEntity("OperateType", eachData["operateType"].ToString(), FieldEntity.FieldTypeEnum.Numer),
                           new FieldEntity("orgDb", orgDb, FieldEntity.FieldTypeEnum.Str),
                           new FieldEntity("toDb", toDb, FieldEntity.FieldTypeEnum.Str),
                           new FieldEntity("CreateTime", DateTime.Now, FieldEntity.FieldTypeEnum.Timestamp),
                           new FieldEntity("LastChangeTime", DateTime.Now, FieldEntity.FieldTypeEnum.Timestamp),
                           new FieldEntity("UniqueColValue", eachData["uniqueColValue"].ToString (), FieldEntity.FieldTypeEnum.Str),
                           new FieldEntity("UniqueColName",eachData["uniqueColName"].ToString (), FieldEntity.FieldTypeEnum.Str)
                       }
                    });
                    //找到映射信息
                    JObject tab = CommonHelper.getMapTabByTableName(tabMapTab, tableName, orgDb);
                    if (tab == null)
                    {
                        LogHelp.WriteLog("数据同步-未找到表：" + tableName + "的映射信息，,文件：" + fi.Name);
                        return;
                    }

                    //处理更新信息
                   ExcuteEntity ex= GetExcuteData(eachData, tab, orgDb, toDb);
                   if (ex != null)
                   {
                       exs.Add(ex);                    
                   }
                }
                //CommonHelper.saveReceiveFileRecord(obj, fi.Name);//存入文件接收记录
            }
            if (exsAsync.Count > 0)
            {
                allTaskList.Add(Task.Factory.StartNew(() =>
                {
                    try
                    {
                        DatabaseHelper.ExecuteDataByEntitys(exsAsync);
                    }
                    catch (Exception ex)
                    {
                        LogHelp.WriteError("接收处理类-存入接收记录异常", ex);
                    }
                }));
            }
            if (exs.Count>0)
                DatabaseHelper.ExecuteDataByEntitys(exs);
            
            //移动文件
            CommonHelper.moveFile(fi, receiveFileBak); 

        }

        /// <summary>
        /// 将每行数据解析成entity执行对象
        /// </summary>
        /// <param name="eachData"></param>
        /// <returns></returns>
        ExcuteEntity GetExcuteData(Dictionary<String, Object> eachData, JObject thisTab, String orgDb, String toDb)
        {
            ExcuteEntity re = new ExcuteEntity();
            re.TableName = eachData["tableName"].ToString();
            Dictionary<String, Object> newData = JsonConvert.DeserializeObject<Dictionary<String, Object>>(eachData["newValue"].ToString()); 
            var uniqueColName = eachData["uniqueColName"].ToString().ToUpper();//唯一列
            String uniqueColValue = eachData["uniqueColValue"].ToString();//此行的唯一值
            if (eachData["operateType"].ToString() == "1")
            {
                re.ExcuteType = ExcuteEntity.ExcuteTypeEnum.insert;
                var excuteFileds=new List<FieldEntity>();
                foreach (String key in newData.Keys) {
                    String[] _key = key.Split('@');
                    Object fieldVal = newData[key];
                    //获取目标数据库的映射字段
                    Dictionary<string, Object> toField = CommonHelper.getToField(thisTab, orgDb, toDb, _key[0]);
                    if (toField == null)
                    {
                        continue;
                    }
                    if (toField["mapType"].ToString() == "str")
                    {
                        _key[0] = toField[toDb].ToString();
                    }
                    if (toField["mapType"].ToString() == "attr")//此处要存入到属性表中
                    {
                        if (fieldVal != null && !string.IsNullOrWhiteSpace(fieldVal.ToString()))
                        {
                            CommonHelper.saveAttrsData((JObject)toField[toDb], uniqueColValue, fieldVal.ToString());
                            continue;
                        }
                    }

                    FieldEntity.FieldTypeEnum fieldType = FieldEntity.FieldTypeEnum.Str;
                    CommonHelper.GetFieldValAndType(_key,ref fieldType,ref fieldVal);
                    excuteFileds.Add(new FieldEntity(_key[0], fieldVal, fieldType));
                }
                re.ExcuteFileds = excuteFileds;

            }else if (eachData["operateType"].ToString() == "2")
            {
                re.ExcuteType = ExcuteEntity.ExcuteTypeEnum.update;
                var excuteFileds = new List<FieldEntity>();
                var whereFileds = new List<FieldEntity>();

                foreach (String key in newData.Keys)
                {
                    String[] _key = key.Split('@');
                    if (_key[0].ToUpper() == eachData["uniqueColName"].ToString().ToUpper()) {
                        continue;
                    }
                    Object fieldVal = newData[key];
                    //获取目标数据库的映射字段
                    Dictionary<string, Object> toField = CommonHelper.getToField(thisTab, orgDb, toDb, _key[0]);
                    if (toField == null)
                    {
                        continue;
                    }
                    if (toField["mapType"].ToString() == "str")
                    {
                        _key[0] = toField[toDb].ToString();
                    }
                    if (toField["mapType"].ToString() == "attr")//此处要存入到属性表中
                    {
                        if (fieldVal != null && !string.IsNullOrWhiteSpace(fieldVal.ToString()))
                        {
                            CommonHelper.saveAttrsData((JObject)toField[toDb], uniqueColValue, fieldVal.ToString());
                            continue;
                        }
                    }

                    FieldEntity.FieldTypeEnum fieldType = FieldEntity.FieldTypeEnum.Str;
                    CommonHelper.GetFieldValAndType(_key, ref fieldType, ref fieldVal);
                    excuteFileds.Add(new FieldEntity(_key[0], fieldVal, fieldType));
                }
                whereFileds.Add(new FieldEntity(eachData["uniqueColName"].ToString(), eachData["uniqueColValue"].ToString(), FieldEntity.FieldTypeEnum.Str));
                re.ExcuteFileds = excuteFileds;
                re.WhereFileds = whereFileds;
                if (excuteFileds.Count == 0) {
                    return null;
                }
            }
            else if (eachData["operateType"].ToString() == "3")
            {
                re.ExcuteType = ExcuteEntity.ExcuteTypeEnum.del;
                var whereFileds = new List<FieldEntity>();
                whereFileds.Add(new FieldEntity(eachData["uniqueColName"].ToString(), eachData["uniqueColValue"].ToString(), FieldEntity.FieldTypeEnum.Str)); 
                re.WhereFileds = whereFileds;
            }


            return re;
        }

        
    }

}
